package com.chinamobile.iot.handler.redis;

import com.alibaba.fastjson.JSON;
import com.chinamobile.iot.config.BrokerConfiguration;
import com.chinamobile.iot.handler.impl.BrokerMonitorImpl;
import com.chinamobile.iot.handler.model.BrokerInfo;
import com.chinamobile.iot.handler.model.SubscriptionMessage;
import com.chinamobile.iot.handler.model.Tree3Client;
import com.chinamobile.iot.mqtt.spi.impl.SubscriptionTree;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by szl on 2017/7/11.
 */
@Component
@ConditionalOnExpression("${mqtt.distribute}")
public class RedisListener {

    private static Logger logger = LoggerFactory.getLogger(RedisListener.class);

    @Autowired
    private BrokerMonitorImpl brokerMonitor;

    @Autowired
    private BrokerConfiguration brokerConfiguration;

    private SubscriptionTree<String> tree2;
    private SubscriptionTree<Tree3Client> tree3;

    /**
     * 0 ： onConnected should ignore all the unsubscribe message
     * 1 ： normal should processor all the message realtime
     */
    private AtomicInteger workMode = new AtomicInteger(0);


    public SubscriptionTree<String> getTree2() {
        return tree2;
    }

    public void setTree2(SubscriptionTree<String> tree) {
        this.tree2 = tree;
    }

    public SubscriptionTree<Tree3Client> getTree3() {
        return tree3;
    }

    public void setTree3(SubscriptionTree<Tree3Client> tree3) {
        this.tree3 = tree3;
    }

    public void receiveMessage(String message) {
        logger.info("receive message from redis channel : {}", message);
        SubscriptionMessage sub = JSON.parseObject(message, SubscriptionMessage.class);
        if (sub != null) {
            switch (sub.getType()) {
                case -1: {
                    switch (workMode.get()) {
                        case 0: {
                            // in onConnected mode， the unsubscribe message is ignored
                            // in normal mode， when the broker discover other brokers send unsubscribe message to themself,
                            // it should publish unsubscribe message in the redis channel again
                        }
                        break;
                        case 1: {
                            // in normal mode, the unsubscribe message is processord
                            // 清除第三课树
                            if (!sub.isCleanSession() && tree3 != null) {
                                Tree3Client client3 = new Tree3Client(sub.getClientId(), sub.getBroker());
                                tree3.removeSubscribe(client3, sub.getFilter());
                            } else {
                                // 清除第二棵订阅树
                                if (sub.getBroker() != null && tree2 != null) {
                                    tree2.removeSubscribe(sub.getBroker(), sub.getFilter());
                                }
                            }
                        }
                        break;
                    }
                }
                break;
                case 1: {
                    // 如果收到的是自己broker发出去的消息，直接忽略
                    if (sub.getBroker() != null && sub.getBroker().equals(brokerConfiguration.getName())) {
                        return;
                    } else {
                        if (sub.isCleanSession()) {
                            if (sub.getBroker() != null) {
                                ConcurrentHashMap<String, BrokerInfo> map = brokerMonitor.getMapBrokerInfo();
                                if (map.containsKey(sub.getBroker()) && tree2 != null) {
                                    tree2.add(sub.getFilter(), sub.getBroker(), sub.getQos());
                                }
                            }
                        } else {
                            if (tree3 != null) {
                                Tree3Client client3 = new Tree3Client(sub.getClientId(), sub.getBroker());
                                tree3.add(sub.getFilter(), client3, sub.getQos());
                            }
                        }
                    }
                }
                break;
                default: {
                }
            }
        }
    }
    public void setNormalMode() {
        workMode.set(1);
    }
}
